package app.dws;

import com.bw.gmall.bean.DiYiZgiErBao;
import com.bw.gmall.utils.MyClickHouseUtil;
import com.bw.gmall.utils.MyKafkaUtil;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;

public class DwsDiErZhiBiao {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("create table action_log( " +
                "    `common` map<string,string>, " +
                "    `action` map<string,string>, " +
                "    `page` map<string,string>, " +
                "    `ts` bigint, " +
                "    `rt` as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)), " +
                "    WATERMARK FOR rt AS rt - INTERVAL '2' SECOND " +
                " ) " + MyKafkaUtil.getKafkaDDL("dwd_traffic_action_log", "dwd_traffic_action_log_1565"));

        tableEnv.executeSql("create table sku_info( " +
                "`after` map<string,string>)" + MyKafkaUtil
                .getKafkaDDL("dwd_sku_info_add", "dwd_sku_info_add_1565"));

        Table table = tableEnv.sqlQuery("select " +
                "`page`['item'] as sku_id," +
                "`common`['uid'] as uid," +
                "`common`['is_new'] as is_new," +
                "`common`['ch'] as ch," +
                "`action`['action_id'] as action_id," +
                "rt " +
                "from action_log " +
                "where `page`['item'] is not null");

        Table sku_info = tableEnv.sqlQuery("select " +
                "`after`['id'] as id," +
                "`after`['sku_name'] as sku_name " +
                "from sku_info");

        tableEnv.createTemporaryView("zho",table);
        tableEnv.createTemporaryView("sname",sku_info);

        Table xiadan = tableEnv.sqlQuery("" +
                "select " +
                "a1.uid," +
                "a1.sku_id," +
                "a1.is_new," +
                "COALESCE(a1.ch,'0') as ch, " +
                "COALESCE(a1.action_id,'0') as action_id, " +
                "COALESCE(a2.sku_name,'0') as sku_name, " +
                "DATE_FORMAT(TO_TIMESTAMP(tt, 'yyyy-MM-dd HH:mm:ss'), 'yyyy-MM-dd') as tt " +
                "from(" +
                "select " +
                "SPLIT_INDEX(sku_id, ',', 0) as sku_id," +
                "is_new," +
                "uid," +
                "ch," +
                "action_id," +
                "DATE_FORMAT(rt, 'yyyy-MM-dd HH:mm:ss') as tt " +
                "from zho )a1 " +
                "left join sname a2 on a1.sku_id=a2.id");

        tableEnv.createTemporaryView("cu_jie",xiadan);

        Table table1 = tableEnv.sqlQuery("select " +
                "uid,tt," +
                "sum(cast(is_new as int)) as is_new," +
                "action_id," +
                "LISTAGG(ch,',') as ch," +
                "LISTAGG(sku_name,',') as sku_name," +
                "sum(if(action_id = 'favor_add',1,0)) as favor_num," +
                "sum(if(action_id = 'trade_add_address',1,0)) as pay_num," +
                "sum(if(action_id = 'cart_add',1,0)) as cart_num," +
                "sum(if(action_id = 'get_coupon',1,0)) as get_coupon_num," +
                "sum(if(action_id = '',1,0)) as fang_num " +
                "from cu_jie " +
                "group by uid,tt,action_id,ch");
        tableEnv.createTemporaryView("er_ban",table1);

        Table table2 = tableEnv.sqlQuery("select " +
                "COALESCE(CONCAT('人气ID',uid,':'),'00') as id,tt," +
                "COALESCE(CONCAT('一位',if(is_new>0,'新客','老客')),'00') as is_new," +
                "case\n" +
                "        when action_id = 'favor_add' then '收藏了以下商品'\n" +
                "        when action_id = 'trade_add_address' then '购买了以下商品'\n" +
                "        when action_id = 'cart_add' then '加购了以下商品'\n" +
                "        when action_id = 'get_coupon' then '领取了以下优惠劵'\n" +
                "        else '访问了您的商品' end as dui," +
                "COALESCE(action_id,'000') as action_id," +
                "CASE \n" +
                "    WHEN INSTR(ch, ',') > 0 \n" +
                "    THEN SUBSTRING(ch, 1, INSTR(ch, ',') - 1)\n" +
                "    ELSE ch\n" +
                "  END as ch," +
                "COALESCE(sku_name,'00') as sku_name," +
                "case\n" +
                "        when action_id = 'favor_add' then '，这是TA过去一年第'\n" +
                "        when action_id = 'trade_add_address' then '，这是TA过去一年第'\n" +
                "        when action_id = 'cart_add' then '，这是TA过去一年第'\n" +
                "        when action_id = 'get_coupon' then '，这是TA过去一年第'\n" +
                "        else '，这是TA过去一年第' end as der," +
                "case\n" +
                "        when action_id = 'favor_add' then favor_num\n" +
                "        when action_id = 'trade_add_address' then pay_num\n" +
                "        when action_id = 'cart_add' then cart_num\n" +
                "        when action_id = 'get_coupon' then get_coupon_num\n" +
                "        else fang_num end as shu," +
                "case\n" +
                "        when action_id = 'favor_add' then '次收藏本店商品！'\n" +
                "        when action_id = 'trade_add_address' then '次购买本店商品！'\n" +
                "        when action_id = 'cart_add' then '次加购本店商品！'\n" +
                "        when action_id = 'get_coupon' then '次领取本店优惠券！'\n" +
                "        else '次访问本店商品！' end as dsa " +
                "from er_ban ");
        tableEnv.createTemporaryView("wan_chu",table2);

        DataStream<Row> rowDataStream = tableEnv.toChangelogStream(table2);
        SingleOutputStreamOperator<DiYiZgiErBao> appendStream = rowDataStream
                .filter(row -> {
                    RowKind kind = row.getKind();
                    // 只保留插入和更新后的数据
                    return kind == RowKind.INSERT || kind == RowKind.UPDATE_AFTER ||
                            kind == RowKind.DELETE || kind == RowKind.UPDATE_BEFORE;
                })
                .map(row -> {

                    return new DiYiZgiErBao(
                            row.getField(0).toString(),
                            row.getField(1).toString(),
                            row.getField(2).toString(),
                            row.getField(3).toString(),
                            row.getField(4).toString(),
                            row.getField(5).toString(),
                            row.getField(6).toString(),
                            row.getField(7).toString(),
                            row.getField(8).toString(),
                            row.getField(9).toString()
                    );
                }, TypeInformation.of(DiYiZgiErBao.class));



        appendStream.print();

        appendStream.addSink(MyClickHouseUtil.
                getSinkFunction("insert into dwd_di_yi_zhou_er values(?,?,?,?,?,?,?,?,?,?)"));


        env.execute();
    }
}
